
package com.web.kdl.config;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.web.kdl.daoV2.matrix.MatrixDao;
import com.web.kdl.entity.matrix.Matrix;
import com.web.kdl.entity.site.Site;
import com.web.kdl.entity.sitewheel.SiteWheel;
import com.web.kdl.service.flywheelDataAnalysis.FlywheelDataAnalysisService;
import com.web.kdl.service.site.SiteService;
import com.web.kdl.service.siteWheel.SiteWheelService;
import com.web.kdl.util.RedisUtils;
import com.web.kdl.util.TopicUtils;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

@Configuration
public class MqttModbusConfig {

    public static final String CHANNEL_NAME_IN = "mqttInputChannel";
    private static final Logger log = LoggerFactory.getLogger(MqttModbusConfig.class);

    @Value("${custom-mqtt.username}")
    private String username;

    @Value("${custom-mqtt.password}")
    private String password;

    @Value("${custom-mqtt.url}")
    private String url;

    @Value("${custom-mqtt.consumer.clientId}")
    private String clientId;

    @Value("${custom-mqtt.consumer.defaultTopic}")
    private String topic;

    //新的部分，java端需要作为生产者向kdl发送数据
    @Value("${custom-mqtt.producer.clientId}")
    private String producerClientId;

    @Value("${mqtt.transactionId}")
    private int localTransactionId;


    @Autowired
    private FlywheelDataAnalysisService flywheelDataAnalysisService;

    @Autowired
    private SiteWheelService siteWheelService;

    @Autowired
    private SiteService siteService;

    @Autowired
    private MatrixDao matrixDao;

    @Autowired
    private RedisUtils redisUtils;



    /**
     * 生产者MQTT客户端
     */
    @Bean(name = "mqttProducerClient")
    public MqttClient mqttProducerClient() throws MqttException {
        // 1. 创建客户端实例
        MqttClient client = new MqttClient(url, producerClientId);

        // 2. 配置连接选项
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setAutomaticReconnect(true); // 自动重连
        options.setCleanSession(true);       // 清理会话

        // 3. 连接到Broker
        client.connect(options);
        return client;
    }

    // //配合主启动类创建的一个配置
    // @Bean
    // public ThreadPoolTaskScheduler taskScheduler() {
    //     ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    //     scheduler.setPoolSize(2);
    //     scheduler.setThreadNamePrefix("mqtt-task-");
    //     scheduler.initialize();
    //     return scheduler;
    // }

    /**
     * MQTT信息通道（消费者）
     *创建消息通道，用于MQTT消息的传递
     * DirectChannel：点对点消息通道
     */
    @Bean(name = CHANNEL_NAME_IN)
    public MessageChannel mqttInputChannel() {
        return new ExecutorChannel(Executors.newFixedThreadPool(20));
    }

    /**
     * MQTT连接器选项
     *
     */
    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true); // false-保持持久会话 true-不保持
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setServerURIs(StringUtils.split(url, ","));
        options.setKeepAliveInterval(60); // 心跳间隔
        options.setAutomaticReconnect(true); // 自动重连
        return options;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions()); // 注入连接配置
        return factory;
    }

    /**
     * 这个Bean的作用是建立MQTT连接，订阅指定主题，
     * 并将接收到的消息转发到 mqttInputChannel 通道，最终由 mqttMessageHandler 进行处理。
     * @return
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter  mqttInboundAdapter() {
        //获取需要订阅的主题
        List<Matrix> matrices = matrixDao.selectList(null);
        String[] topics = matrices.stream().map(matrix -> TopicUtils.getSubscribeTopic(matrix.getTopic())).toArray(String[]::new);

//        创建MQTT消息入站适配器
//        MqttPahoMessageDrivenChannelAdapter adapter =
//                new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(), topic);
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(), topics);
        // 创建消息转换器
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        // 将消息转换成字节数组
        converter.setPayloadAsBytes(true);
        // 设置消息转换器
        adapter.setConverter(converter);
        adapter.setQos(1);// 0 1 2
        // 设置输出通道
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
@Bean
@ServiceActivator(inputChannel = CHANNEL_NAME_IN)
public MessageHandler mqttMessageHandler() {
    return message -> {
        try {
            // 记录 时间
            long startTime = System.currentTimeMillis();
            Object payload = message.getPayload();
            String hexResponse;
            //获取订阅主题
            String topic = message.getHeaders().get("mqtt_receivedTopic").toString().split("/")[1];

            // 1. 解析 payload（支持 String 或 byte[]）
            if (payload instanceof String) {
                hexResponse = ((String) payload).trim();
            } else if (payload instanceof byte[]) {
                hexResponse = bytesToHex((byte[]) payload);
            } else {
                System.err.println("不支持的消息类型: " + payload.getClass());
                return;
            }

            // 2. 转为 byte[]
            byte[] response = hexToBytes(hexResponse);
            if (response.length < 9) {
                System.err.println("报文过短");
                return;
            }

            // 3. 提取关键字段

            int transactionId = (response[0] << 8) | (response[1] & 0xFF);
            int protocolId    = (response[2] << 8) | (response[3] & 0xFF);
            int length        = (response[4] << 8) | (response[5] & 0xFF);
            int flyWheelSerial       = response[6] & 0xFF;
            int funcCode      = response[7] & 0xFF;
            int byteCount     = response[8] & 0xFF; //后续数据部分的字节数
            if(transactionId!=localTransactionId) return;

            // 4.  成功识别：是哪台设备？
            int registerNumber = byteCount / 2;


            // 5. 校验
            if (protocolId != 0) {
                System.err.println("非标准 Modbus 协议");
                return;
            }
            if (funcCode == 0x83) {
                log.error("错误响应，异常码={},错误码={}，错误响应数据: {},",response[8],funcCode,hexResponse);
                return;
            }
            if(funcCode == 6){
                log.info("写入寄存器指令");
                return;
            }



            // 6. 提取寄存器数据
            short[] registers = new short[byteCount / 2];
            for (int i = 0; i < registers.length; i++) {
                int offset = 9 + i * 2;
                registers[i] = (short) ((response[offset] << 8) | (response[offset + 1] & 0xFF));
            }

            System.out.println("寄存器数据: " + Arrays.toString(registers));

            //阵列-飞轮1储能分站点
//            Matrix matrix = matrixDao.selectById(matrixId);
            Matrix matrix = matrixDao.selectOne(new LambdaQueryWrapper<Matrix>().eq(Matrix::getTopic,topic ));

            log.info("✅收到设备响应，主题：{},飞轮序号:{},寄存器数量:{}", topic, flyWheelSerial, registerNumber);
            String flywheelName = "flywheel"+flyWheelSerial;
            //7.拆分处理数据并保存到数据库中
            if (registerNumber == 98){//总控的信息
                Site site = siteService.getSiteById(matrix.getSiteId());
                flywheelDataAnalysisService.flywheel98(response, "总控制信息", site.getSumSite(),matrix.getMatrixName() , "flywheel"+flyWheelSerial);
            }else{
                Site site = siteService.getSiteById(matrix.getSiteId());
                if (registerNumber == 100){ //一台飞轮180个寄存器，分两次读取先读取100个（含时间数据）
                    flywheelDataAnalysisService.flywheel100(response, flywheelName, site.getSumSite(),matrix.getMatrixName(), "flywheel"+flyWheelSerial,matrix.getId(),flyWheelSerial);
                }else if (registerNumber == 80){//读取一台飞轮的另外80个寄存器数据
                    flywheelDataAnalysisService.flywheel80(response, flywheelName, site.getSumSite(),matrix.getMatrixName(),"flywheel"+flyWheelSerial,matrix.getId(),flyWheelSerial);
                }
            }
            long endTime = System.currentTimeMillis();
            log.info("MQTT消息处理结束，耗时: {} ms", (endTime - startTime));

        } catch (Exception e) {
            System.err.println("解析 Modbus 响应失败: " + e.getMessage());
            e.printStackTrace();
        }
    };
}

    // 工具方法：Hex 字符串转 byte[]
    private byte[] hexToBytes(String hex) {
        hex = hex.replaceAll("\\s+", ""); // 去空格
        int len = hex.length();
        byte[] data = new byte[len / 2];
        for (int i = 0; i < len; i += 2) {
            data[i / 2] = (byte) ((Character.digit(hex.charAt(i), 16) << 4)
                    + Character.digit(hex.charAt(i + 1), 16));
        }
        return data;
    }

    // 工具方法：byte[] 转 Hex 字符串
    private String bytesToHex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bytes) {
            sb.append(String.format("%02X ", b));
        }
        return sb.toString().trim();
    }
}
